package org.qnit

import org.apache.spark.sql.SparkSession
import org.qnit.util.{MysqlUtil, SparkUtil}

/**
 * @Description:
 * @Author: zhuxianyu
 * @Date: 2023/10/10 21:08
 * @Version: 1.0
 */
object Data2Mysql {

  val tableNames = Array(
    "user_milvus_mapping",
    "dws_user_cnt",
    "dws_expert_logbook_cnt",
    "dws_expert_logbook_yymm_cnt",
    "dws_label_cnt",
    "dws_area_label"
  )

  def main(args: Array[String]): Unit = {
    val env = args(0)
    println(s"env=$env")
    if (!SparkUtil.verifyEnv(env))
      System.exit(1)
    val spark: SparkSession = SparkUtil.initSparkSession(env, "Data2Mysql")

    tableNames.foreach(tableName => {
      println(s"==================== $tableName ====================")
      tableName match {
        case "user_milvus_mapping" =>
          val df = spark.table("dws.dws_user_embedding").select("milvus_key", "user_id").distinct()
          MysqlUtil.save(df, tableName)
        case _ =>
          val df = spark.table(s"dws.$tableName")
          MysqlUtil.save(df, tableName)
      }
    })

    spark.stop()
  }

}
